package cn.jly.boot.kafka.service.impl;

import cn.jly.boot.kafka.service.KafkaProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @PackageName cn.jly.boot.kafka.service.impl
 * @ClassName KafkaProducerImpl
 * @Description
 * @Author 姬岚洋
 * @Date 2021/1/10 下午 9:22
 */
@Service
public class KafkaProducerImpl implements KafkaProducer {

    /**
     * logger
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProducerImpl.class);
    /**
     * error message
     */
    private static final String ERROR_MESSAGE = "send data to kafka error : {}";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * // 声明事务：后面报错消息不会发出去
     * kafkaTemplate.executeInTransaction(operations -> {
     * operations.send("topic1","test executeInTransaction");
     * throw new RuntimeException("fail");
     * });
     * <p>
     * // 不声明事务：后面报错但前面消息已经发送成功了
     * kafkaTemplate.send("topic1","test executeInTransaction");
     * throw new RuntimeException("fail");
     *
     * @param topic   主题
     * @param message 消息
     */
    @Override
    public void sendMessage(String topic, String message) {
        try {
            kafkaTemplate.send(topic, message);
        } catch (Exception e) {
            LOGGER.error(ERROR_MESSAGE, e.getMessage());
            e.printStackTrace();
        }
    }

    @Override
    public void sendMessageWithCallback(String topic, String message) {
        try {
            kafkaTemplate.send(topic, message).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onFailure(Throwable throwable) {
                    LOGGER.error(ERROR_MESSAGE, throwable.getMessage());
                }

                @Override
                public void onSuccess(SendResult<String, String> result) {
                    final String sendTopic = result.getRecordMetadata().topic();
                    final long offset = result.getRecordMetadata().offset();
                    final int partition = result.getRecordMetadata().partition();
                    final String value = result.getProducerRecord().value();
                    System.out.printf("消息发送成功 -> topic: %s, offset: %s, partition: %s, value: %s %n", sendTopic, offset, partition, value);
                }
            });
        } catch (Exception e) {
            LOGGER.error(ERROR_MESSAGE, e.getMessage());
            e.printStackTrace();
        }
    }
}
